package com.dib.neo.bkfs.fs;

import com.google.common.collect.Lists;
import io.netty.buffer.ByteBuf;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;

public class BKFSClient implements FSClient {

  public static final String BKFS_CONF = "/.conf";
  public static final String BKFS_INOGEN = BKFS_CONF+"/inogen";
  final byte[] ledgerPassword = "alice&elsa".getBytes();

  final String connectString;
  final CuratorFramework client;
  final DistributedAtomicLong atomicInoGen;
  final BookKeeper bkc;
  final PathChildrenCache pathChildrenCache;

  public BKFSClient(String connectString) throws IOException {
    try {
      this.connectString = connectString;
      bkc = new BookKeeper(connectString);
      client = CuratorFrameworkFactory.builder().namespace("bkfs")
          .retryPolicy(new RetryNTimes(10, 5000))
          .connectString(connectString).build();
      client.start();
      atomicInoGen = new DistributedAtomicLong(client, BKFS_INOGEN, new RetryNTimes(10, 5000));
      pathChildrenCache = new PathChildrenCache(client, "/", true);
      pathChildrenCache.start();

    }catch (Exception e){
      throw toIOException(e);
    }
  }

  public CuratorFramework getClient() {
    return client;
  }

  public void  addListener(PathChildrenCacheListener listener){
    pathChildrenCache.getListenable().addListener(listener);
  }

  public void  removeListener(PathChildrenCacheListener listener){
    pathChildrenCache.getListenable().removeListener(listener);
  }

  IOException toIOException(Exception e) {
    if (e instanceof IOException) {
      return (IOException) e;
    }
    return new IOException(e);
  }



  @Override
  public void close() throws IOException {
    try {
      client.close();
      if (bkc != null) {
        bkc.close();
      }

    } catch (Exception e) {
      throw toIOException(e);
    }
  }


  @Override
  public boolean touch(String path) throws IOException {
    return touch(path,0);
  }

  @Override
  public boolean touch(String path, int blkSize) throws IOException {
    if (!exists(path)) {
      String parent = Paths.get(path).getParent().toString();
      if (!exists(parent)) {
        mkdirs(parent);
      }
      try {
        AtomicValue<Long> ino = atomicInoGen.increment();
        if (ino.succeeded()) {
          FileInfo file = new FileInfo(path);
          file.getAttributes().regularFile().setIno(ino.postValue());
          checkBlockSize(blkSize, file);
          client.create().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
              .forPath(path, file.data());
          return true;
        }
      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new FileAlreadyExistsException(path);
    }
    return false;
  }

  private void checkBlockSize(int blkSize, FileInfo file) {
    if (blkSize >= 128) {
      file.getAttributes().setBlockSize(blkSize);
    }
  }

  @Override
  public List<String> ls(String path) throws IOException {
    if (exists(path)) {
      try {
        List<String> names = Lists.newArrayList();
        FileInfo file = this.getFile(path);
        if (file.getAttributes().isDirectory()) {

          List<String> children = client.getChildren().forPath(path);
          for (String child : children) {
            names.add(Paths.get(child).toFile().getName());
          }
        }
        return names;
      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public void rm(String path) throws IOException {
    if (exists(path)) {
      try {
        FileInfo file = this.getFile(path);
        if (file.getAttributes().isDirectory()) {

          List<String> children = client.getChildren().forPath(path);
          if (children != null) {
            for (String child : children) {
              if (child != null) {
                rm(child);
              }
            }
          }
        } else if (file.getAttributes().isRegularFile()) {
          for (Long lId : file.getBlocks()) {
            freeData(lId);
          }
          client.delete().forPath(path);
        } else {
          client.delete().forPath(path);
        }
      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public void mv(String path, String target) throws IOException {
    if (exists(path)) {
      try {
        FileInfo file = this.getFile(path);
        String parent = Paths.get(target).getParent().toString();
        if (!exists(parent)) {
          mkdirs(parent);
        }
        long now = System.currentTimeMillis();
        client.create().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
            .forPath(target, file.data());

        client.delete().forPath(path);
      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public void copy(String path, String target) throws IOException {

  }

  private void freeData(Long lId) {
    try {
      bkc.deleteLedger(lId);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }


  @Override
  public void mkdir(String path) throws IOException {
    mkdir(path,0);
  }

  @Override
  public void mkdirs(String path) throws IOException {
    mkdirs(path,0);
  }

  @Override
  public void mkdir(String path, int blkSize) throws IOException {
    if (!exists(path)) {
      try {
        String parent = Paths.get(path).getParent().toString();
        if (!exists(parent)) {
          throw new NoSuchFileException(parent);
        }
        AtomicValue<Long> ino = atomicInoGen.increment();
        if (ino.succeeded()) {
          FileInfo file = new FileInfo(path);

          long now = System.currentTimeMillis();
          file.getAttributes().directory().setIno(ino.postValue());
          checkBlockSize(blkSize, file);
          client.create().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
              .forPath(path, file.data());
        }


      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new FileAlreadyExistsException(path);
    }
  }

  @Override
  public void mkdirs(String path, int blkSize) throws IOException {
    if (!exists(path)) {
      try {
        Path parent = Paths.get(path).getParent();
        if (parent != null && !exists(parent.toString())) {
          mkdirs(parent.toString());
        }
        AtomicValue<Long> ino = atomicInoGen.increment();
        if (ino.succeeded()) {
          long now = System.currentTimeMillis();
          FileInfo file = new FileInfo(path);
          file.getAttributes().directory().setIno(ino.postValue());
          checkBlockSize(blkSize, file);
          client.create().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
              .forPath(path, file.data());
        }
      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new FileAlreadyExistsException(path);
    }
  }

  @Override
  public boolean exists(String path) {
    try {
      try {
        client.sync().forPath(path);
        Stat exists = client.checkExists().forPath(path);
        if (exists != null) {
          return true;
        }
      } catch (Exception e) {
        throw toIOException(e);
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
    return false;
  }


  @Override
  public FileInfo getFile(String path) throws IOException {
    if (exists(path)) {
      try {
        Stat stat = new Stat();
        //client.sync().forPath(path);

        byte[] data = client.getData().storingStatIn(stat)
            .forPath(path);

        FileInfo file = new FileInfo(path);
        file.syncData(stat,data);
        return file;
      } catch (Exception e) {
        throw toIOException(e);
      }
    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public void truncate(String path,long size) throws IOException {
    if (exists(path)) {
      try {
        FileInfo file = this.getFile(path);
        List<Long> dels = Lists.newArrayList( file.getBlocks());
        file.truncate(size);
        dels.removeAll(file.getBlocks());
        for(Long lId:dels){
          freeData(lId);
        }
        client.setData().withVersion(file.getVersion()).forPath(file.getPath(), file.data());

      } catch (Exception e) {
        throw toIOException(e);
      }

    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public int read(String path,long offset, ByteBuffer dst) throws IOException {
    if (exists(path)) {
      try {
        FileInfo file = this.getFile(path);

        return read(file,dst,offset);
      } catch (Exception e) {
        throw toIOException(e);
      }

    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public int write(String path,long offset,  ByteBuffer src) throws IOException {
    if (exists(path)) {
      try {
        FileInfo file = this.getFile(path);
        return write(file,src,offset);
      } catch (Exception e) {
        throw toIOException(e);
      }

    } else {
      throw new NoSuchFileException(path);
    }
  }

  @Override
  public int read(String path, long offset) throws IOException {
    return 0;
  }


  private int write(FileInfo file, ByteBuffer src, long offset) throws IOException {
    try {
      int length = src.limit();
      int start = (int) (offset / file.getAttributes().getBlockSize());
      int end = (int) Math.ceil((0.0 + offset + length) / file.getAttributes().getBlockSize());
      int twLen = 0;
      List<Long> dels = new ArrayList<>();
      for (int i = start; i < end; i++) {
        int _offset = 0;
        if (i == start) {
          _offset = (int) (offset % file.getAttributes().getBlockSize());
        }
        int remaining = length - twLen;
        int wLen = Math.min(file.getAttributes().getBlockSize() - _offset, remaining);
        //System.out.println("_offset = " + _offset);
        //System.out.println("remaining = " + remaining);
        //System.out.println("wLen = " + wLen);
        if (i < file.getBlockCount()) {
          boolean changed = false;
          long oldId = file.getBlock(i);
          LedgerHandle oldlh = bkc.openLedger(oldId, BookKeeper.DigestType.MAC, ledgerPassword);

          LedgerEntry entry = oldlh.readLastEntry();
          ByteBuf oldbuf = entry.getEntryBuffer().copy();
          oldlh.close();
          if(oldbuf.nioBuffer().limit()==wLen){
            for (int index = 0; index < wLen; index++) {
              int pos = index + _offset;
              byte old = oldbuf.getByte(pos);
              byte val = src.get(twLen + index);
              if (old != val) {
                changed = true;
                oldbuf.setByte(pos, val);
              }
            }
          }else{
            changed = true;
          }

          if (changed) {
            LedgerHandle lh = bkc.createLedger(BookKeeper.DigestType.MAC, ledgerPassword);
            lh.addEntry(oldbuf.array());
            lh.close();
            file.setBlock(i, lh.getId());
            dels.add(oldId);
          }
        } else {
          LedgerHandle lh = bkc.createLedger(BookKeeper.DigestType.MAC, ledgerPassword);
          lh.addEntry(src.array(), twLen, wLen);

          lh.close();
          file.setBlock(i, lh.getId());
        }
        twLen += wLen;

      }
      if ((offset + length) > file.getAttributes().size()) {
        file.getAttributes().setSize((offset + length));
      }
      long now = System.currentTimeMillis();
      client.setData().withVersion(file.getVersion()).forPath(file.getPath(), file.data());
      for (Long lId : dels) {
        //System.out.println("delete deleteLedger lId = " + lId);
        freeData(lId);
      }
      dels.clear();
    } catch (Exception e) {
      throw toIOException(e);
    }
    return src.limit();
  }



  private int read(FileInfo file,ByteBuffer dst,long offset)
      throws IOException, BKException, InterruptedException {
    int length = dst.capacity();
    int start = (int) (offset / file.getAttributes().getBlockSize());
    int end = (int) Math.ceil((0.0 + offset + length) / file.getAttributes().getBlockSize());
    int trLen = 0;
    for (int i = start; (i < end) && (i < file.getBlockCount()); i++) {
      Long ledgerId = file.getBlock(i);
      LedgerHandle lh = bkc.openLedger(ledgerId, BookKeeper.DigestType.MAC, ledgerPassword);

      LedgerEntry entry = lh.readLastEntry();
      lh.close();

      ByteBuffer buf = entry.getEntryBuffer().copy().nioBuffer();
      int offset0 = 0;
      if (i == start) {
        offset0 = (int) (offset % file.getAttributes().getBlockSize());
      }

      int remaining = length - trLen;
      int rLen = Math.min(remaining,buf.limit()-offset0);
      if(rLen>0){
        dst.put(buf.array(), offset0, rLen);
        trLen += rLen;
      }
    }

    return trLen;
  }


}
